/**
 * @file   Producer.hpp
 * @author Pedro Jimenez
 * @date   Mon Nov 10 12:31:03 2008
 * 
 * @brief  Support classes for thread safe circular buffer and threads
 * 
 * 
 */

// Comparison of bounded buffers based on different containers.

// Copyright (c) 2003-2008 Jan Gaspar

// Use, modification, and distribution is subject to the Boost Software
// License, Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)
#ifndef _C_PRODUCER_HPP_
#define _C_PRODUCER_HPP_

#define BOOST_CB_DISABLE_DEBUG

#include <boost/circular_buffer.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/thread.hpp>
#include <boost/progress.hpp>
#include <boost/bind.hpp>
#include <deque>
#include <list>
#include <string>
#include <iostream>

/**
 * @brief This class implements a thread-safe circular buffer, FIFO style. 
 * 
 * This circular buffer is intended to help passing objects in a typical \b Producer - 
 * \b Consumer architecture, where the Producer and the Consumer live at different threads of execution.
 * It implements two fifos, where always the same objects rotate from producer to consumer, and back to producer, avoiding extra initializations. 
 * 
 * One FIFO passes filled objects from the producer to the consumer, 
 * and another fifo passed the already used elements from the consumer to the producer, again. 
 * Basically, the Producer is collecting data, which put in a buffer, and passes this buffer to the Consumer, which will process this data. 
 * This circular buffer stores up to \c m_capacity objects of type \b T. All the objects must be initialized first prior to push them into the buffer. After that, the Producer only fills up objects, after passes them to the Consumer by calling *get_next_free*
 * The consumer asks for a filled buffer by calling get_next_used, use its content,
 * and sent it back to the Producer by calling again get_next_used. The Producer can now destroy the content of this last buffer and fill it up again with new data.
 * This way, each object of the buffer is only initialized once, and the objects rotates from the producer to the consumer, and to the producer again.
 */
template <class T>
class circular_buffer {
private:
    /// buffer which stores all the objects in the circular_buffer
    T * m_container;  
    /// pointer to the next avaible buffer to the producer, to be filled up, returned by \c get_last_free()
    T * p_free; 
    /// pointer to the last used buffer by the producer. Used internally to know when the buffer is full
    T * p_free_old; 
    ///pointer to the next filled up buffer to the consumer, returned by \c get_last_used()
    T * p_used; 
    ///pointer to the last used buffer by the consumer
    T * p_used_old;
    /// last element of the \c m_container buffer of the circular_buffer
    T * p_fin_container;
    /// Maximum number of elements in the circular_buffer. Size of the \c m_container buffer
    int m_capacity;
    /// Actual number of elements in the circular_buffer.
    int m_size;
    
    boost::mutex m_mutex;
    boost::condition_variable m_not_full;
    boost::condition_variable m_not_empty;
    int m_queue_used;
    bool m_quit_buffer;

    circular_buffer(const circular_buffer<T>&);              // Disabled copy constructor
    circular_buffer& operator = (const circular_buffer<T>&); // Disabled assign operator

public:

    /// This defines the data type of the buffer, to be used on other non-member functions
    typedef T value_type;

    /**
     * @brief Default constructor
     * @param capacity indicates the maximum number of elements in the circular buffer, and thus the size of the \c m_container. 
     * If capacity != 0, it initializes m_container with capacity elements.
     */
    circular_buffer(int capacity=0) {
	m_size=0;
	m_queue_used=0;
	m_capacity  = capacity;
	m_quit_buffer = 0;
	m_container = 0;
	if (m_capacity)
	    init(m_capacity);
    }
    /**
     * @brief Destructor. 
     * 
     * deletes the memory assigned to \c m_container. 
     */
    ~circular_buffer()
    {
	delete []m_container;
    }

    /**
     * @brief Initializes the circular buffer. If it was initializes before, it does nothing. <B> It doesn't change the size of the circular_buffer container </B>
     * @param capacity indicates the maximum number of elements in the circular buffer, and thus the size of the m_container. 
     * If capacity != 0, it initializes m_container with capacity elements.
     */
    void init(int capacityin)
    {
	if (m_container) 
	    return;
	m_size = 0;
	m_quit_buffer =0;
	m_queue_used = 0;
	m_capacity = capacityin;
	//create the container
	m_container = new T[m_capacity];
	p_free = m_container;
	p_used = m_container;
	p_free_old = m_container;
	p_used_old = m_container;
	p_fin_container = m_container;
    }

    /**
     * @brief Includes a new element to the buffer
     * @param pItem is the new element <b>to be copied</b> to the m_container of the circular buffer.
     * @return true if the container of the circular_buffer is not jet full, and so more elements can be inserted in.
     */
    bool push(T &pItem) {
	if (m_size == m_capacity)
	    return false;
	p_fin_container = &m_container[m_size++];
	*p_fin_container = pItem;
	p_used_old = p_fin_container;
	return true;
    }
    
    ///Removes one element from the circular buffer
    T* pop(void)
    {
	if (m_size== 0)
	    return NULL;
	if (p_used_old == p_fin_container)
	    p_used_old --;
	else if (p_used == p_fin_container)
	    p_used --;

	if (p_free_old == p_fin_container)
	    p_free_old --;
	else if (p_free == p_fin_container)
	    p_free --;

	p_fin_container--;
	m_size--;
	return p_fin_container+1;
    }

    /// Indicates that there's space for more elements
    bool is_not_full() const {	return m_size < m_capacity; }
    /// Indicates that there's any element in the container
    bool is_not_empty()const {	return m_size > 0; }

    /** 
     * \brief Indicates that there's filled available buffers to the Consumer
     * 
     * A call to \c get_next_used() will block if this functions returns false, until the Producer fills up a new object and calls \c get_next_free().
     * @return 
     */
    bool there_are_used() const { return p_used != p_free_old; }
    /// 
    /** 
     * \brief Indicates that there's free available buffers to the Producer
     * 
     * A call to \c get_next_free() will block if this functions returns false, until the Consumer uses an object and calls \c get_next_free() to free the buffer.
     * @return true if there is any free available buffer.
     */
    bool there_are_free() const { return p_used_old != p_free; }
    
    /**
     * @brief returns the actual available filled buffer, to be used by the consumer. 
     * @return the last buffer returned buffer by \c get_next_used().
     */
    T* get_last_used() { return p_used; }
    /**
     * @brief returns the actual available empty buffer, to be used by the Producer. 
     * @return the last buffer returned buffer by \c get_next_free().
     */
    T* get_last_free() { return p_free; }

    /**
     * \brief returns the next available filled buffer, to be used by the consumer. 
     *
     * This function frees the actual buffer being used by the consumer and makes it available for the producer, 
     * and returns the next available buffer with data from the producer. If there's no buffers with data available, 
     * the call blocks until the Producer calls get_next_free().
     * @return the actual available buffer with data to be used by the consumer. Also returned by \c get_last_used().
     * @return NULL if there is no more buffers and the circular_buffer is stopped, so never is going to be.
     */
    T* get_next_used() {
	boost::unique_lock<boost::mutex> lock(m_mutex);
	while (!there_are_used()) {
	    if (m_quit_buffer)
	    {
		lock.unlock();
		return NULL;
	    }
	    m_not_empty.wait(lock);
	}
	p_used_old = p_used;
	if (p_used == p_fin_container)
	    p_used = m_container;
	else
	    p_used ++;
	m_queue_used --;
	lock.unlock();

	m_not_full.notify_one();
	return get_last_used();
    }

    /**
     * \brief returns the next available empty buffer, to be used by the producer. 
     *
     * This function push into the filled fifo the actual buffer filled up by the producer and makes it available for the consumer, 
     * and returns the next available empty buffer. If there's no empty buffers, 
     * the call blocks until the consumer calls get_next_used().
     * @return the actual available empty buffer to be used by the producer. Also returned by \c get_last_free().
     * @return NULL if there is no more buffers and the circular_buffer is stopped, so never is going to be.
     */
    T* get_next_free() {
	boost::unique_lock<boost::mutex> lock(m_mutex);
	while(!there_are_free()) {
	    m_not_full.wait(lock);
	    /*if (m_quit_buffer)
	    {
		lock.unlock();
		return NULL;
		}*/
	}
	p_free_old = p_free;
	if (p_free == p_fin_container)
	    p_free = m_container;
	else
	    p_free ++;
	m_queue_used ++;
	lock.unlock();
	m_not_empty.notify_one();
	return get_last_free();
    }

    /** 
     * @brief Unblocks the calls to \c get_next_free() and \c get_next_used(), if any, and stops the circular_buffer.
     *
     * If any thread of execution is block in a call to \c get_next_free() of \c get_next_used(), the call is unblocked, and the functions returns NULL,
     * indicating that there's no more available buffer.
     * 
     */
    void cancel(){
	m_quit_buffer = true;
	m_not_empty.notify_one();
	m_not_full.notify_one();
	boost::unique_lock<boost::mutex> unlock(m_mutex);
	
    }
    /** 
     * \brief beauty print of the stay of the buffer
     * 
     * The template class T necessarily must have the member function \c get1() and \c get2(), which would be used to print data from the class.
     * @param inuse the current buffer being used by the consumer
     * @param infill the current buffer being used by the producer
     */
    void print(T *inuse, T* infill)
    {
	boost::unique_lock<boost::mutex> lock(m_mutex);
	int i;
	int p;
	p = p_free - m_container;
	printf("\n");
	for (i=0; i<p; i++)
	    printf("      ");
	printf("Producer | ");
	infill->print();
	printf("\n       ");
	for (i=0; i<m_size; i++)
	    printf("[%2d], ", m_container[i].get1());
	for (; i<m_capacity; i++)
	    printf("[  ], ");
	printf("\n       ");
	for (i=0; i<m_size; i++)
	    printf("[%2d], ", m_container[i].get2());
	for (; i<m_capacity; i++)
	    printf("[  ], ");
	printf("size %d/%d\n", m_size, m_capacity);
	p = p_used - m_container;
	for (i=0; i<p; i++)
	    printf("      ");
	printf("Consumer | ");
	inuse->print("\n");
    }
	
    int capacity() const {return m_capacity;}
    int size() const { return m_size;}
    int size_queue_free ()  { return m_size - m_queue_used; }
    int size_queue_used ()  { return m_queue_used; }
};
    


/** 
 * \brief This class implements a typical Producer-thread / Consumer process, with a circular_buffer passing data between them
 * 
 * The provides the appropriates \c virtual methods to launch a thread which will contain the \b Producer, and calls to pass and get data through a \c\b circular_buffer from the Producer to the consumer and vice. 
 */
template <class T>
class CProducer {
protected:
    /// Thread safe circular_buffer to pass data between the Producer and the consumer
    circular_buffer<T> m_container;
    bool m_quit_thread;
    /// Producer thread object
    boost::thread *m_thread;
public:

    /** 
     * @brief Only initializes the circular_buffer
     * 
     * @param capacityin Size of the circular_buffer
     */
    CProducer(int capacityin)
	: m_container(capacityin) 
    {
	m_quit_thread=0;
    }
    /** 
     * \brief Stops the Producer thread and the circular_buffer
     * 
     */
    ~CProducer()
    {
	stop();
    }
    /** 
     * \brief Starts the Producer thread
     * @return false if the Producer thread could not be Started
     */
    virtual bool start(void)
    {
	m_quit_thread=0;
	//start the CProducer thread
	m_thread = new boost::thread (boost::ref(*this));
	return m_thread != NULL;
    }
    /** 
     * \brief Producer thread function.
     * 
     * This function has an infinity loop which consecutive calls \c Producer() to get empty buffer from the circular_buffer and then calls the 
     * <b> pure virtual function \c producer_callback() </b> to process data and fillup the buffer for the consumer. 
     */
    virtual void operator()()//=0 ;
    {
	T* p_empty;
	//thread
	p_empty = producer();
        while ( producer_callback(p_empty)) {
	    p_empty = producer();
	    boost::this_thread::yield();
	}
	m_container.cancel();
    }
    /** 
     * \brief Pure virtual function which will implements the Producer main task.
     * 
     * This function is called each loop of the producer main thread, with a new empty buffer which has to be filled up by this function.
     * The CProducer class will automatically, unless overridden, collect the a new free or empty buffer, call \c producer_callback(T* item) with this buffer, and 
     * queue the buffer in the circular_buffer, making it available to the consumer.
     * @param infill the free buffer to be filled up by the function.
     * @return false to stop the producer. True to go on with the loop, and send the buffer to the consumer
     */    
    virtual bool producer_callback(T* infill)=0;
    /** 
     * \brief calls to \c circular_buffer::get_next_free() function to get the next empty or free buffer.
     * 
     * This function is available in case the derived class wants to override it.
     * @return the buffer returned by \b circular_buffer::get_next_free()
     */
    virtual T* producer()
    {
            return m_container.get_next_free();
    };
    /** 
     * \brief Stops the Producer thread, the calls to \c producer_callback(), and the circular_buffer.
     * 
     */
    virtual void stop(void)
    {
	if (m_quit_thread)
	    return;
	m_quit_thread = true;
	m_container.cancel();
	m_thread->interrupt();
	m_thread->join();
    }
    /** 
     * \brief calls to \c circular_buffer::get_next_used() function to get the next filled buffer.
     * 
     * This function is available in case the derived class wants to override it.
     * @return the buffer returned by \b circular_buffer::get_next_used()
     */
    virtual T* consumer() 
    {
	return m_container.get_next_used();
    }
    /** 
     * \brief calls to \c circular_buffer::push() function to insert a new item to the member \c m_container circular_buffer.
     * 
     * This function is available in case the derived class wants to override it.
     * @return false if the circular_buffer is full.
     * @param pItem Element to be inserted in the member circular_buffer.
     */
    virtual bool push_buffer(T &pItem)
    {
	if (!m_container.push(pItem))
	    return false;
	return m_container.is_not_empty();
    }
    /** 
     * \brief calls to \c circular_buffer::pop() function to remove an item from the member \c m_container circular_buffer.
     * 
     * This function is available in case the derived class wants to override it.
     * @return the removed item;
     * @return NULL is empty.
     */
    virtual T* pop_buffer(void)
    {
	return m_container.pop();
    }

    virtual inline bool is_quit() {return m_quit_thread;}
    int size() const { return m_container.size();}
    int size_queue_free ()  { return m_container.size_queue_free();}
    int size_queue_used ()  { return m_container.size_queue_used();}

    virtual T*get_last_free() { return m_container.get_last_free(); }
    virtual T*get_last_used() { return m_container.get_last_used(); }

    virtual void print(T *a, T * b) {return m_container.print(a, b);}

};

#endif

//  LocalWords:  fifos fifo param Destructor bool
